package com.atguigu.chapter07;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/9 10:16
 */
public class Flink27_State_BroadcastState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<String> inputDS = env.socketTextStream("localhost", 9999);
        DataStreamSource<String> controlDS = env.socketTextStream("localhost", 8888);

        // TODO 1.把 其中一条流（控制流） 广播出去
        MapStateDescriptor<String, String> broadcastStateDesc = new MapStateDescriptor<>("broadcast-state", Types.STRING, Types.STRING);
        BroadcastStream<String> controlBS = controlDS.broadcast(broadcastStateDesc);

        // TODO 2.把 另一条流 和 广播流 关联起来（connect）
        BroadcastConnectedStream<String, String> inputBCS = inputDS.connect(controlBS);

        // TODO 3.调用 process
        inputBCS
                .process(new BroadcastProcessFunction<String, String, String>() {
                    /**
                     * 输入流的处理逻辑： 在里面获取只读的广播状态，进行对应处理
                     * @param value
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                        // TODO 5.通过上下文获取广播状态，取出里面的值(只读的，不能修改)
                        ReadOnlyBroadcastState<String, String> broadcastState = ctx.getBroadcastState(broadcastStateDesc);
                        String aSwitch = broadcastState.get("switch");
                        if ("1".equals(aSwitch)) {
                            out.collect("切换到1的逻辑.....");
                        } else if ("0".equals(aSwitch)) {
                            out.collect("切换到0的逻辑.....");
                        } else {
                            out.collect("切换到其他逻辑....");
                        }


                    }

                    /**
                     * 处理广播流的数据： 这里主要定义 ，什么数据往广播状态存
                     * @param value
                     * @param ctx
                     * @param out
                     * @throws Exception
                     */
                    @Override
                    public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {
                        // TODO 4. 通过上下文获取广播状态,往广播状态里存数据
                        BroadcastState<String, String> broadcastState = ctx.getBroadcastState(broadcastStateDesc);
                        broadcastState.put("switch", value);
                    }
                })
                .print();


        env.execute();
    }
}
/*
    广播状态应用场景：
        1）广播流最好是： 数据量小、更新频率低的场景
        2）可以用于： 动态配置更新、 类似开关的切换执行逻辑、  动态规则匹配 ...
 */